package io.xxx.aurora.client.produce

import com.alibaba.fastjson.JSON
import io.netty.bootstrap.Bootstrap
import io.netty.channel.Channel
import io.netty.channel.ChannelInitializer
import io.netty.channel.ChannelOption
import io.netty.channel.epoll.EpollEventLoopGroup
import io.netty.channel.epoll.EpollServerSocketChannel
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.string.StringDecoder
import io.netty.handler.codec.string.StringEncoder
import io.netty.util.concurrent.Future
import io.netty.util.concurrent.GenericFutureListener
import io.xxx.aurora.client.netty.EchoInboundHandler
import io.xxx.aurora.client.netty.Sender
import io.xxx.aurora.common.util.OperatingSystem
import org.slf4j.LoggerFactory
import java.net.InetSocketAddress
import java.util.*

interface Producer {

    fun send(record: ProducerRecord)
}

class AuroraProducer(props: Properties) : Producer {

    private val channel: Channel

    init {
        val servers = props.getProperty("bootstrap.servers")
        val split = servers.split(",")[0].split(":")
        val host = split[0]
        val port = split[1].toInt()
        val group = NioEventLoopGroup()
        val b = Bootstrap()
        b.group(group)
            .channel(NioSocketChannel::class.java)
            .handler(object : ChannelInitializer<SocketChannel>() {
                override fun initChannel(ch: SocketChannel) {
                    ch.pipeline()
                        .addLast(StringEncoder())
                        .addLast(StringDecoder())
                        .addLast(EchoInboundHandler())
                }
            })
        val f = b.connect(host, port).sync()
        f.addListener {
            if (!it.isSuccess) {
                f.cause().printStackTrace()
                group.shutdownGracefully()
            }
        }
        channel = f.channel()
    }

    override fun send(record: ProducerRecord) {
        fun partitionCountForTopic(topic: String): Int {
            return 0
        }

        val partitionCount = partitionCountForTopic(record.topic)
        val partition = record.partition ?: 0
        val timestamp = record.timestamp ?: System.currentTimeMillis()
        channel.writeAndFlush(JSON.toJSONString(record))
    }

    companion object {
        private val log = LoggerFactory.getLogger(AuroraProducer::class.java)
    }
}